/**
 * Copyright 2006-2014 handu.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.jianbing.rocketmq;


import com.aliyun.openservices.ons.api.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 发送消息
 *
 * @author  Han, Tixiang
 * @since 2015-01-16
 */
public class AliyunRocketMQSink extends AbstractSink implements Configurable {
    private Log logger = LogFactory.getLog(AliyunRocketMQSink.class);
    public static final String TOPIC_CONFIG = "topic";
    public static final String TOPIC_DEFAULT = "FLUME_ROCKETMQ";
    /**
     * Tags配置项，如：a1.sinks.s1.tags=Tag1,Tag2
     */
    public static final String TAG_CONFIG = "tag";
    public static final String TAG_DEFAULT = "";
    private static final Logger LOG = LoggerFactory.getLogger(AliyunRocketMQSink.class);

    private String topic;
    private String tag;
    private Producer producer;

    @Override
    public void configure(Context context) {
        // 获取配置项
        topic = context.getString(TOPIC_CONFIG, TOPIC_DEFAULT);
        tag = context.getString(TAG_CONFIG, TAG_DEFAULT);
        // 初始化Producer
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ProducerId, context.getString(PropertyKeyConst.ProducerId));
        properties.put(PropertyKeyConst.AccessKey, context.getString(PropertyKeyConst.AccessKey));
        properties.put(PropertyKeyConst.SecretKey, context.getString(PropertyKeyConst.SecretKey));
        properties.put(PropertyKeyConst.ONSAddr, context.getString(PropertyKeyConst.ONSAddr));
        producer = ONSFactory.createProducer(properties);
        //在发送消息前，必须调用start方法来启动Producer，只需调用一次即可。
        producer.start();
    }

    @Override
    public Status process() throws EventDeliveryException {
        Channel channel = getChannel();
        Transaction tx = channel.getTransaction();
        try {
            tx.begin();
            Event event = channel.take();
            if (event == null || event.getBody() == null || event.getBody().length == 0) {
                tx.commit();
                return Status.READY;
            }
            // 发送消息
            SendResult sendResult = producer.send(new Message(topic, tag, event.getBody()));
            logger.info("send ok");
            tx.commit();
            return Status.READY;
        } catch (Exception e) {
            LOG.error("RocketMQSink send message exception", e);
            try {
                tx.rollback();
                return Status.BACKOFF;
            } catch (Exception e2) {
                LOG.error("Rollback exception", e2);
            }
            return Status.BACKOFF;
        } finally {
            tx.close();
        }
    }

    @Override
    public synchronized void start() {
        super.start();
    }

    @Override
    public synchronized void stop() {
        // 停止Producer
        producer.shutdown();
        super.stop();
    }
}
